/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package com.wider.foundation.concurrency;

import com.wider.foundation.DLogger;
import com.wider.foundation.Named;
import com.wider.foundation.TimeLimited;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import javax.naming.SizeLimitExceededException;

/**
 * This class defines an Executor Pool (pool of threads) which supports:
 * 1. Adaptive pool size, and
 * 2. Aged request.
 *
 * @author  William Choi <avextk@gmail.com>
 * @see     CountedBlockingQueue
 * @see     AggressiveWorker
 */
public class AdaptiveAgedExecutorPool implements Named {
    private static final int DEFAULT_POOL_SIZE_INCS = 8;
    private static final int DEFAULT_MAX_POOL_SIZE = 8;
    private static final int DEFAULT_POLLING_PERIOD = 2000;
    
    private boolean     isShut;
    private boolean     isTerm;
    private String      name;
    private int         max_queue_len;
    private int         max_pool_size;
    private int         cur_pool_size;
    private int         max_increment;
    private int         cur_increment;
    private int         cur_watermark;
    
    private DLogger                         logger;
    private HashSet<AggressiveWorker>       workers;
    private CountedBlockingQueue<Runnable>  queue;


    /**
     * Class constructor for creating the AdaptiveAgedExecutorPool instance.
     * In order to enable this pool, user must call the <code>init()</code> method after calling the constructor.
     * 
     * @param given     The given name of this instance.
     */
    public AdaptiveAgedExecutorPool (String given) {
        max_queue_len   = 0;
        max_pool_size   = 0;
        max_increment   = 0;
        cur_increment   = 0;
        cur_pool_size   = 0;
        cur_watermark   = 0;
        name    = given;
        isShut  = true;
        isTerm  = false;
        logger  = DLogger.getInstance(this);

        logger.debug("Constructed.");
    }


    /**
     * The simplest init'er using DEFAULT_MAX_POOL_SIZE and DEFAULT_POOL_SIZE_INCS
     * @param max_pendings      The max queue size
     */
    public void init(int max_pendings){
        init(max_pendings, DEFAULT_MAX_POOL_SIZE, DEFAULT_POOL_SIZE_INCS);
    }

    /**
     * Simple init'er using the DEFAULT_POOL_SIZE_INCS (num of adaptive increments)
     * @param max_pendings      The max queue size
     * @param max_poolsize      The max executor pool size
     */
    public void init(int max_pendings, int max_poolsize){
        init(max_pendings, max_poolsize, DEFAULT_POOL_SIZE_INCS);
    }

    /**
     * Detailed init'er which user can specify their preferred num of increments (adaptive pool size)
     * @param max_pendings      The max queue size
     * @param max_poolsize      The max executor pool size
     * @param poolsize_incs     The max number of increments
     */
    public void init(int max_pendings, int max_poolsize, int poolsize_incs) {
        if (!isShut || isTerm)
            throw new IllegalStateException("Illegal State to init' this instance.");
        
        max_queue_len   = max_pendings;
        max_pool_size   = max_poolsize;
        max_increment   = poolsize_incs;

        queue   = new CountedBlockingQueue<Runnable>(max_queue_len);
        workers = new HashSet<AggressiveWorker>();
        
        //Skip to pool size >= 2
        for (int start_pool_start = 0; start_pool_start < 1;){
            cur_increment++;
            start_pool_start   = (int) Math.round(Math.exp(cur_increment) * max_pool_size / Math.exp(max_increment));
        }

        logger.trace("entering...");
        try {
            expand(); //expand to next incremental
        } catch (SizeLimitExceededException ex) {
            logger.warn(ex);
        }
        isShut = false;
        logger.trace("exited.");
    }


    /**
     * Method for user to submit an Runnable task to the pool
     * @param task  The task to be submitted
     * @throws NullPointerException
     * @throws IllegalStateException
     */
    public void submit(Runnable task) throws NullPointerException, IllegalStateException {
        //first add the task to queue
        //this operation may throw IllegalStateException for full queue,
        //or NullPointerException for null task
        do{
            if ((task instanceof Poison) || (!isShut && !isTerm)){
                queue.add(task);
                String temp = (task instanceof Poison) ? "Poison" : "Task";
                logger.debug(temp + " is submitted.");
            }else{
                logger.debug("task is ignored.");
                break;
            }

            //Check the current queue depth to see if we need to expand the thread pool
            if (queue.size() > cur_watermark){
                logger.debug("Queue reaches watermark, perform expansion...");
                try {
                    expand();
                } catch (SizeLimitExceededException ex) {
                    logger.warn(ex);
                    break;
                }
                logger.debug("expanded");
            }
        }while(false);
    }


    /**
     * Method for user to cancel the submitted task which may be pending in the queue or being processed by the Executor.
     * @param task  The submitted task.
     * @throws IllegalStateException
     * @deprecated
     */
    public void cancel(Runnable task) throws IllegalStateException {
        if (isShut || isTerm)
            throw new IllegalStateException("Service is shut down or terminiated.");

        //First, check if this task exists in the waiting queue.
        logger.trace("entering...");
        if (!queue.remove(task)){
            if (task instanceof Cancellable){
                ((Cancellable) task).cancel();
            }
        }
        logger.trace("exited.");
    }


    /**
     * Method for awaiting the termination of the pool. This method will wait all Executor thread to join back.
     * @param timeout   This awaiting will block the user up to timeout in unit. 0 means infinit timeout.
     * @param unit      The time unit of the timeout value
     * @return  True if successful await; false otherwise.
     */
    public boolean awaitTermination(long timeout, TimeUnit unit){
        if (!isShut)
            throw new IllegalStateException("Should be shutdown before await for termination.");

        boolean ret = true;

        do{
            //Wait for all threads (consumers) to park to the queue
            TimeLimited ticket  = new TimeLimited(unit.toMillis(timeout));
            long    waitingtime = unit.toMillis(timeout);
            while (!(ticket.isExpired()) && (queue.getBlockingCount() < cur_pool_size)){
                synchronized(queue){
                    try {
                        logger.debug("try waiting for:::: " + waitingtime + ". current blocking count = " + queue.getBlockingCount());
                        queue.wait(waitingtime);
                        waitingtime = ticket.timeleft();
                        waitingtime = (-1 == waitingtime) ? DEFAULT_POLLING_PERIOD : waitingtime;
                    } catch (InterruptedException ex) {
                        logger.warn(ex);
                    }
                }
            }
            if ((ticket.isExpired()) && (queue.getBlockingCount() < cur_pool_size)){
                logger.debug("Timedout while waiting the workers to park back.");
                ret = false;
                break;
            }


            //All the consumers were parked back on the queue, send STOP signal
            for (Stoppable worker : workers){
                worker.stop();
            }

            //Broadcast Poison (fake Runnable) to all workers
            for (int i = 0; i < cur_pool_size; i++){
                submit(new Poison());
            }

            //Wait for all workers to join back
            for (Joinable worker : workers){
                try {
                    worker.join();
                } catch (InterruptedException ex) {
                    logger.warn(ex);
                }
            }
        }while(false);

        isTerm = ret;

        return ret;
    }

    /**
     * Calling this method to shutdown accepting new request before calling <code>awaitTermination()</code>
     */
    public void shutdown(){
        isShut = true;
    }

    /**
     * Check if this pool is shut down
     * @return  True if so; false otherwise.
     */
    public boolean isShutdown(){
        return isShut;
    }

    /**
     * Check if this pool is terminated (successful return from <code>awaitTermination()</code>)
     * @return True if so; false otherwise.
     */
    public boolean isTerminated(){
        return isTerm;
    }

    /**
     * Expand the pool to the next increment. If the pool size has reached the Max bound, SizeLimit exception will be thrown
     * @throws SizeLimitExceededException
     */
    private void expand() throws SizeLimitExceededException{
        if ((cur_increment == max_increment) || (cur_pool_size == max_pool_size))
            throw new SizeLimitExceededException("Reached the max pool size, pool couldn't be expanded.");
        
        cur_increment++;
        int new_pool_size = (int) Math.round(Math.exp(cur_increment) * max_pool_size / Math.exp(max_increment));
        int new_watermark = (int) Math.round(Math.exp(cur_increment) * max_queue_len / Math.exp(max_increment));

        logger.debug("Expansion:: pool size: from " + cur_pool_size + " to " + new_pool_size + ", watermark = " + new_watermark + ". max pool size = " + max_pool_size);
        for ( ; cur_pool_size < new_pool_size; cur_pool_size++){
            AggressiveWorker worker = new AggressiveWorker("Worker" + cur_pool_size);
            workers.add(worker);
            worker.setSource(queue);
            worker.start();
        }
        logger.warn("NEW POOL SIZE = " + cur_pool_size);
        
        cur_watermark = new_watermark;
        
    }

    @Override
    public String getName() {
        return name;
    }

    public int getMaxQueueSize(){
        return max_queue_len;
    }

    public int getMaxPoolSize(){
        return max_pool_size;
    }

    public int getPendingCount(){
        return queue.size();
    }

    public int getActiveCount(){
        return cur_pool_size - queue.getBlockingCount();
    }
}
